package org.etlt.load;

import org.etlt.extract.Extractor;
import org.etlt.job.JobContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

public abstract class Loader implements Comparable<Loader>, AutoCloseable {

    public static final String COUNTER = "${counter.interval}";
    private String name;

    private final LoaderSetting setting;

    private long loadedRecordCount;

    private LoaderStatus status = LoaderStatus.INITED;

    protected Loader(LoaderSetting setting) {
        this.setting = setting;
    }

    public <T extends LoaderSetting> T getSetting() {
        return (T) this.setting;
    }

    public long getLoadedRecordCount(){
        return this.loadedRecordCount;
    }

    protected void count(int loadedNumber){
        this.loadedRecordCount += loadedNumber;
    }

    /**
     * the default interval is 1000
     * @param context
     * @return
     */
    protected int getCounterInterval(JobContext context){
        try {
            String intervalAsString = context.getPropertyValue(COUNTER);
            int interval = Integer.parseInt(intervalAsString);
            return interval;
        }catch (Exception e){
            return 10000;
        }
    }

    public abstract void preLoad(JobContext context);

    /**
     * load the data to target
     *
     * @param context
     */
    public abstract void load(JobContext context);

    public abstract void doFinish(JobContext context);

    public void close(AutoCloseable... resources) {
        for (AutoCloseable resource : resources) {
            try {
                if (resource != null)
                    resource.close();
            } catch (Exception e) {
                // do nothing
            }
        }
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    /**
     * resolve columns from first extractor, append all user defined columns after auto resolved column settings，<br>
     * if name is same, user defined column setting would override auto resolved column setting
     *
     * @param context
     */
    protected void resolveColumns(JobContext context) {
        List<ColumnSetting> columnSettings = new ArrayList<>();
        if (this.setting.isAutoResolve()) {
            Extractor extractor = context.getExtractor(this.setting.getExtractor());
            List<String> extractorColumns = extractor.getColumns();
            for (String columnName : extractorColumns) {
                if(!this.setting.getRemoveColumns().contains(columnName))
                    columnSettings.add(new ColumnSetting(columnName,
                            this.setting.getExtractor().trim() + "." + columnName
                            ));
            }
            for (ColumnSetting autoResolvedColumnSetting : columnSettings) {
                for (ColumnSetting userDefinedColumnSetting : this.setting.getColumns()) {
                    if (userDefinedColumnSetting.getName().equals(autoResolvedColumnSetting.getName())) {
                        autoResolvedColumnSetting.setExpression(userDefinedColumnSetting.getExpression());
                    }
                }
            }
            appendColumns(columnSettings);
        }
    }

    /**
     * append and override
     *
     * @param columnSettings
     */
    private void appendColumns(List<ColumnSetting> columnSettings) {
        List<ColumnSetting> allColumns = new ArrayList<>(columnSettings);
        List<ColumnSetting> userDefined = this.setting.getColumns();
        userDefined.forEach(setting -> {
            if (allColumns.contains(setting)) {
                int index = allColumns.indexOf(setting);
                allColumns.set(index, setting);
            }else
                allColumns.add(setting);
        });
        this.setting.setColumns(allColumns);
    }

    @Override
    public int compareTo(Loader anotherLoader) {
        return this.getName().compareTo(anotherLoader.getName());
    }

    @Override
    public boolean equals(Object anotherLoader) {
        if (this == anotherLoader) return true;
        if (anotherLoader == null || getClass() != anotherLoader.getClass()) return false;
        Loader loader = (Loader) anotherLoader;
        return Objects.equals(name, loader.name);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name);
    }

    public LoaderStatus getStatus() {
        return status;
    }

    public void setStatus(LoaderStatus status) {
        this.status = status;
    }

    @Override
    public String toString() {
        return "Loader{" +
                "name= " + name +
                ", status = " + status +
                "}";
    }

}
